package com.tqz.rc.client.beat;

import com.tqz.rc.api.BeatInfo;
import com.tqz.rc.api.Instance;
import com.tqz.rc.client.naming.NamingHttpClientProxy;
import com.tqz.rc.common.constant.*;
import com.tqz.rc.common.lifecycle.Closeable;
import com.tqz.rc.common.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <p>客户端处理心跳。
 *
 * @author tianqingzhao
 * @since 2021/8/30 17:24
 */
public class BeatReactor implements Closeable {

    private static final Logger log = LoggerFactory.getLogger(BeatReactor.class);

    private Map<String, BeatInfo> dom2Beat = new ConcurrentHashMap<>(10);

    private NamingHttpClientProxy namingClientProxy;

    private ScheduledExecutorService executorService;

    public BeatReactor(NamingHttpClientProxy namingClientProxy) {
        this.namingClientProxy = namingClientProxy;
        executorService = new ScheduledThreadPoolExecutor(1);
    }

    /**
     * 构建心跳对象
     *
     * @param serviceName 服务名称
     * @param instance    实例信息
     * @return
     */
    public BeatInfo buildBeatInfo(String serviceName, Instance instance) {
        BeatInfo beatInfo = new BeatInfo();
        beatInfo.setNamespaceId(instance.getNamespaceId());
        beatInfo.setServiceName(serviceName);
        beatInfo.setClusterName(instance.getClusterName());
        beatInfo.setIp(instance.getIp());
        beatInfo.setPort(instance.getPort());
        beatInfo.setPeriod(DefaultValueConstant.DEFAULT_HEART_BEAT_INTERVAL);
        return beatInfo;
    }

    /**
     * 添加心跳检查任务，也就是开启一个定时任务向服务端发送该实例还健康的请求。每5秒钟发送一次。
     *
     * @param beatInfo 心跳包
     */
    public void addBeatInfo(BeatInfo beatInfo) {
        String key = buildKey(beatInfo.getServiceName(), beatInfo.getIp(), beatInfo.getPort());

        BeatInfo existBeat = null;
        // 同一个服务只能有一个心跳续约任务，防止心跳任务重复
        if ((existBeat = dom2Beat.remove(key)) != null) {
            existBeat.setStopped(true);
        }
        dom2Beat.put(key, beatInfo);

        executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    }

    private String buildKey(String serviceName, String ip, int port) {
        return serviceName + Constants.NAMING_INSTANCE_ID_SPLITTER + ip + Constants.NAMING_INSTANCE_ID_SPLITTER + port;
    }

    class BeatTask implements Runnable {

        private BeatInfo beatInfo;

        public BeatTask(BeatInfo beatInfo) {
            this.beatInfo = beatInfo;
        }

        @Override
        public void run() {

            if (beatInfo.isStopped()) {
                return;
            }

            if (log.isDebugEnabled()) {
                log.debug("客户端调用服务续约时间：{}", DateUtil.getCurDateTime());
            }

            // 构建请求的参数
            Map<String, String> params = buildSendBeatInfoParams(beatInfo);

            namingClientProxy.reqApi(UrlConstant.INSTANCE_BEAT_URL, params, HttpMethodConstant.PUT_METHOD);

            // 死循环调用自己，默认也就是每5秒钟调用一次
            // 这里采用 `executorService.schedule(BastTrsk, nextTime, TimeUnit.MILLISECONDS);` 而非传统的定时任务，
            // 这样做不需要考虑定时任务的时间间隔与任务执行耗时对比，比如传统的定时器可能存在这种问题：当前任务还没有结束，下次任务又开始了(就需要额外引入锁或者分布式任务)
            executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
        }

        private Map<String, String> buildSendBeatInfoParams(BeatInfo beatInfo) {
            Map<String, String> result = new HashMap<>();
            result.put(FieldConstant.NAMESPACE_ID_PARAM, beatInfo.getNamespaceId());
            result.put(FieldConstant.SERVICE_NAME_PARAM, beatInfo.getServiceName());
            result.put(FieldConstant.CLUSTER_NAME_PARAM, beatInfo.getClusterName());
            result.put(FieldConstant.IP_PARAM, beatInfo.getIp());
            result.put(FieldConstant.PORT_PARAM, String.valueOf(beatInfo.getPort()));
            return result;
        }
    }

    @Override
    public void shutdown() {
        String className = this.getClass().getName();
        log.info("{} do shutdown begin.", className);
        executorService.shutdown();

        int retry = 3;

        while(retry > 0) {
            retry--;

            try {
                if (executorService.awaitTermination(100, TimeUnit.MILLISECONDS)) {
                    return;
                }
            } catch (InterruptedException exception) {
                executorService.shutdownNow();
                Thread.interrupted();
            } catch (Throwable throwable) {
                log.error("停止BeatReactor任务失败，错误信息：", throwable);
            }
        }

        executorService.shutdownNow();

        log.info("{} do shutdown stop.", className);
    }
}
